--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit 1788aea0c2d17db5ae2363191d877c44d7582440
Parents : a8687a0
Author : Sudo-Ivan <ivan@quad4.io>
Signature : Signature validation error
Date : 2026-01-14T17:59:22-06:00
Add tests for crash recovery, emergency mode, and lifecycle management
- Introduced new tests for heuristic analysis in crash recovery to validate error diagnosis.
- Added tests for emergency mode to ensure memory concurrency handling with the in-memory database.
- Created a comprehensive suite for lifecycle management, including database provider disposal and identity context teardown to prevent memory leaks.
- Improved existing tests for WebAudioBridge to verify proper event loop handling and resource cleanup.
Changes
6 files changed, 381 insertions(+), 4 deletions(-)
Diff
diff --git a/tests/backend/test_crash_recovery.py b/tests/backend/test_crash_recovery.py
index cab6877a..02e6fd0e 100644
--- a/tests/backend/test_crash_recovery.py
+++ b/tests/backend/test_crash_recovery.py
@@ -119,8 +119,28 @@ class TestCrashRecovery(unittest.TestCase):
self.assertIn("!!! APPLICATION CRASH DETECTED !!!", report)
self.assertIn("Type: ValueError", report)
self.assertIn("Message: Simulated error for testing", report)
+ self.assertIn("Probabilistic Root Cause Analysis:", report)
self.assertIn("Recovery Suggestions:", report)
+ def test_heuristic_analysis_sqlite(self):
+ exc_type = type("OperationalError", (Exception,), {})
+ exc_type.__name__ = "sqlite3.OperationalError"
+ exc_value = Exception("no such table: config")
+ diagnosis = {"db_type": "memory"}
+
+ causes = self.recovery._analyze_cause(exc_type, exc_value, diagnosis)
+ self.assertTrue(len(causes) > 0)
+ self.assertEqual(causes[0]["description"], "In-Memory Database Sync Failure")
+
+ def test_heuristic_analysis_asyncio(self):
+ exc_type = RuntimeError
+ exc_value = RuntimeError("no current event loop")
+ diagnosis = {}
+
+ causes = self.recovery._analyze_cause(exc_type, exc_value, diagnosis)
+ self.assertTrue(len(causes) > 0)
+ self.assertIn("Asynchronous Initialization", causes[0]["description"])
+
if __name__ == "__main__":
unittest.main()
diff --git a/tests/backend/test_emergency_mode.py b/tests/backend/test_emergency_mode.py
index d98538d5..420b54ef 100644
--- a/tests/backend/test_emergency_mode.py
+++ b/tests/backend/test_emergency_mode.py
@@ -1,12 +1,14 @@
import os
import shutil
import tempfile
+import threading
from unittest.mock import MagicMock, patch
import pytest
import RNS
from meshchatx.meshchat import ReticulumMeshChat
+from meshchatx.src.backend.database.provider import DatabaseProvider
@pytest.fixture
@@ -221,3 +223,45 @@ def test_normal_mode_startup_logic(mock_rns, temp_dir):
# Verify IntegrityManager.save_manifest WAS called
assert mock_integrity_instance.save_manifest.call_count == 1
+
+
+def test_emergency_mode_memory_concurrency(mock_rns, temp_dir):
+ """Verify that :memory: database connection is shared across threads."""
+ # Reset singleton
+ DatabaseProvider._instance = None
+
+ with (
+ patch(
+ "meshchatx.src.backend.identity_context.IdentityContext.start_background_threads",
+ ),
+ patch("meshchatx.src.backend.identity_context.create_lxmf_router"),
+ patch("meshchatx.meshchat.WebAudioBridge"),
+ patch("meshchatx.meshchat.memory_log_handler"),
+ ):
+ app = ReticulumMeshChat(
+ identity=mock_rns["id_instance"],
+ storage_dir=temp_dir,
+ reticulum_config_dir=temp_dir,
+ emergency=True,
+ )
+
+ ctx = app.current_context
+ provider = ctx.database.provider
+ assert provider.db_path == ":memory:"
+
+ # Set value in main thread
+ test_name = "Emergency Worker"
+ ctx.config.display_name.set(test_name)
+
+ # Simulate another thread by swapping thread-local storage
+ original_local = provider._local
+ provider._local = threading.local()
+
+ try:
+ # Should still return the SAME connection object because of the fix
+ val = ctx.config.display_name.get()
+ assert val == test_name
+ finally:
+ provider._local = original_local
+
+ DatabaseProvider._instance = None
diff --git a/tests/backend/test_lifecycle.py b/tests/backend/test_lifecycle.py
new file mode 100644
index 00000000..1cf589d2
--- /dev/null
+++ b/tests/backend/test_lifecycle.py
@@ -0,0 +1,187 @@
+import gc
+import os
+import shutil
+import sqlite3
+import tempfile
+from unittest.mock import MagicMock, patch
+
+import pytest
+import RNS
+from hypothesis import given
+from hypothesis import strategies as st
+
+from meshchatx.src.backend.database.provider import DatabaseProvider
+from meshchatx.src.backend.identity_context import IdentityContext
+from meshchatx.src.backend.web_audio_bridge import WebAudioBridge
+
+
+def test_database_provider_disposal():
+ """Test that DatabaseProvider correctly closes connections."""
+ db_path = os.path.join(tempfile.gettempdir(), "test_disposal.db")
+ if os.path.exists(db_path):
+ os.remove(db_path)
+
+ # Ensure any existing singleton is cleared
+ DatabaseProvider._instance = None
+
+ try:
+ provider = DatabaseProvider.get_instance(db_path)
+ conn = provider.connection
+ assert isinstance(conn, sqlite3.Connection)
+
+ # Test close()
+ provider.close()
+ with pytest.raises(sqlite3.ProgrammingError, match="closed database"):
+ conn.execute("SELECT 1")
+
+ # Re-open
+ conn2 = provider.connection
+ assert conn2 is not conn
+
+ # Test close_all()
+ provider.close_all()
+ with pytest.raises(sqlite3.ProgrammingError, match="closed database"):
+ conn2.execute("SELECT 1")
+
+ finally:
+ if os.path.exists(db_path):
+ try:
+ os.remove(db_path)
+ except Exception:
+ pass
+ DatabaseProvider._instance = None
+
+
+def test_web_audio_bridge_disposal():
+ """Test that WebAudioBridge correctly manages clients and cleanup."""
+ mock_tele_mgr = MagicMock()
+ mock_config_mgr = MagicMock()
+ bridge = WebAudioBridge(mock_tele_mgr, mock_config_mgr)
+
+ mock_client = MagicMock()
+ mock_tele = MagicMock()
+ mock_tele.active_call = True
+ mock_tele_mgr.telephone = mock_tele
+
+ with (
+ patch("meshchatx.src.backend.web_audio_bridge.WebAudioSource"),
+ patch("meshchatx.src.backend.web_audio_bridge.WebAudioSink"),
+ patch("meshchatx.src.backend.web_audio_bridge.Tee"),
+ patch("meshchatx.src.backend.web_audio_bridge.Pipeline"),
+ ):
+ bridge.attach_client(mock_client)
+ assert mock_client in bridge.clients
+
+ bridge.on_call_ended()
+ assert bridge.tx_source is None
+ assert bridge.rx_sink is None
+ assert len(bridge.clients) == 0
+
+
+def test_identity_context_teardown_completeness():
+ """Verify that teardown cleans up all major components."""
+ mock_identity = MagicMock(spec=RNS.Identity)
+ mock_identity.hash = b"test_hash_32_bytes_long_01234567"
+ mock_identity.get_private_key.return_value = b"mock_pk"
+
+ mock_app = MagicMock()
+ mock_app.storage_dir = tempfile.mkdtemp()
+
+ with (
+ patch("meshchatx.src.backend.identity_context.Database"),
+ patch("meshchatx.src.backend.identity_context.ConfigManager"),
+ patch("meshchatx.src.backend.identity_context.create_lxmf_router"),
+ patch("meshchatx.src.backend.identity_context.IntegrityManager"),
+ patch(
+ "meshchatx.src.backend.identity_context.AutoPropagationManager",
+ ),
+ ):
+ context = IdentityContext(mock_identity, mock_app)
+ context.start_background_threads = MagicMock()
+ context.register_announce_handlers = MagicMock()
+
+ context.setup()
+
+ # Capture instances
+ db_instance = context.database
+ auto_prop_instance = context.auto_propagation_manager
+
+ context.teardown()
+
+ # Verify component cleanup
+ db_instance._checkpoint_and_close.assert_called()
+ auto_prop_instance.stop.assert_called()
+ assert context.running is False
+
+
+def test_identity_context_memory_leak():
+ """Verify that IdentityContext can be garbage collected after teardown."""
+ mock_identity = MagicMock(spec=RNS.Identity)
+ mock_identity.hash = b"leak_test_hash_32_bytes_long_012"
+ mock_identity.get_private_key.return_value = b"mock_pk"
+ mock_app = MagicMock()
+ mock_app.storage_dir = tempfile.mkdtemp()
+
+ import weakref
+
+ # We use a list to store the ref so we can access it after the function scope
+ leak_ref = []
+
+ def run_lifecycle():
+ with (
+ patch("meshchatx.src.backend.identity_context.Database"),
+ patch("meshchatx.src.backend.identity_context.ConfigManager"),
+ patch("meshchatx.src.backend.identity_context.create_lxmf_router"),
+ patch("meshchatx.src.backend.identity_context.IntegrityManager"),
+ patch("RNS.Transport"),
+ ):
+ context = IdentityContext(mock_identity, mock_app)
+ context.start_background_threads = MagicMock()
+ context.register_announce_handlers = MagicMock()
+ context.setup()
+ context.teardown()
+
+ leak_ref.append(weakref.ref(context))
+ # End of with block and function scope should clear 'context'
+
+ run_lifecycle()
+
+ # Break any potential cycles in the app mock which might have captured the context
+ mock_app.reset_mock()
+
+ # Multiple collection rounds
+ for _ in range(5):
+ gc.collect()
+
+ # Check if it was collected
+ assert leak_ref[0]() is None, "IdentityContext was not garbage collected"
+
+
+@given(st.integers(min_value=1, max_value=3))
+def test_identity_context_repeated_lifecycle(n):
+ """Fuzz the lifecycle by repeating setup/teardown multiple times with new instances."""
+ mock_identity = MagicMock(spec=RNS.Identity)
+ mock_identity.hash = b"fuzz_hash_32_bytes_long_01234567"
+ mock_identity.get_private_key.return_value = b"mock_pk"
+
+ mock_app = MagicMock()
+ mock_app.storage_dir = tempfile.mkdtemp()
+
+ with (
+ patch("meshchatx.src.backend.identity_context.Database"),
+ patch("meshchatx.src.backend.identity_context.ConfigManager"),
+ patch("meshchatx.src.backend.identity_context.create_lxmf_router"),
+ patch("meshchatx.src.backend.identity_context.IntegrityManager"),
+ patch("RNS.Transport"),
+ ):
+ for _ in range(n):
+ context = IdentityContext(mock_identity, mock_app)
+ context.start_background_threads = MagicMock()
+ context.register_announce_handlers = MagicMock()
+ context.setup()
+ assert context.running is True
+ context.teardown()
+ assert context.running is False
+
+ if os.path.exists(mock_app.storage_dir):
+ shutil.rmtree(mock_app.storage_dir)
diff --git a/tests/backend/test_property_based.py b/tests/backend/test_property_based.py
index d80cc3b6..bfc5336d 100644
--- a/tests/backend/test_property_based.py
+++ b/tests/backend/test_property_based.py
@@ -374,9 +374,9 @@ def test_identity_restore_base32_robustness(data):
@given(
st.lists(
st.text(min_size=1).filter(
- lambda x: "\n" not in x and x.strip() and x.isalnum()
- )
- )
+ lambda x: "\n" not in x and x.strip() and x.isalnum(),
+ ),
+ ),
)
def test_markdown_renderer_list_rendering(items):
if not items:
diff --git a/tests/backend/test_rns_config_management.py b/tests/backend/test_rns_config_management.py
new file mode 100644
index 00000000..e4077e69
--- /dev/null
+++ b/tests/backend/test_rns_config_management.py
@@ -0,0 +1,87 @@
+import os
+import shutil
+import tempfile
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from meshchatx.meshchat import ReticulumMeshChat
+
+
+@pytest.fixture
+def temp_dir():
+ dir_path = tempfile.mkdtemp()
+ yield dir_path
+ shutil.rmtree(dir_path)
+
+
+@pytest.fixture
+def mock_rns():
+ with (
+ patch("RNS.Reticulum") as mock_reticulum,
+ patch("RNS.Transport"),
+ patch("RNS.Identity") as mock_identity,
+ ):
+ mock_id_instance = MagicMock()
+ mock_id_instance.hash = b"test_hash_32_bytes_long_01234567"
+ mock_id_instance.get_private_key.return_value = b"test_key"
+ mock_identity.return_value = mock_id_instance
+
+ yield {"Reticulum": mock_reticulum, "id_instance": mock_id_instance}
+
+
+def test_rns_config_auto_creation(mock_rns, temp_dir):
+ """Test that Reticulum config is created if it does not exist."""
+ config_dir = os.path.join(temp_dir, ".reticulum")
+ config_file = os.path.join(config_dir, "config")
+
+ # Ensure it doesn't exist
+ assert not os.path.exists(config_file)
+
+ with (
+ patch("meshchatx.meshchat.IdentityContext"),
+ patch("meshchatx.meshchat.WebAudioBridge"),
+ patch("meshchatx.meshchat.memory_log_handler"),
+ ):
+ ReticulumMeshChat(
+ identity=mock_rns["id_instance"],
+ storage_dir=temp_dir,
+ reticulum_config_dir=config_dir,
+ )
+
+ # Method should have been called during init -> setup_identity
+ assert os.path.exists(config_file)
+
+ with open(config_file) as f:
+ content = f.read()
+ assert "[reticulum]" in content
+ assert "[interfaces]" in content
+ assert "enable_transport = False" in content
+
+
+def test_rns_config_repair_if_invalid(mock_rns, temp_dir):
+ """Test that Reticulum config is recreated if it is invalid/corrupt."""
+ config_dir = os.path.join(temp_dir, ".reticulum")
+ os.makedirs(config_dir, exist_ok=True)
+ config_file = os.path.join(config_dir, "config")
+
+ # Create a "corrupt" config
+ with open(config_file, "w") as f:
+ f.write("this is not a valid rns config")
+
+ with (
+ patch("meshchatx.meshchat.IdentityContext"),
+ patch("meshchatx.meshchat.WebAudioBridge"),
+ patch("meshchatx.meshchat.memory_log_handler"),
+ ):
+ ReticulumMeshChat(
+ identity=mock_rns["id_instance"],
+ storage_dir=temp_dir,
+ reticulum_config_dir=config_dir,
+ )
+
+ with open(config_file) as f:
+ content = f.read()
+ # Should have been repaired
+ assert "[reticulum]" in content
+ assert "[interfaces]" in content
diff --git a/tests/backend/test_web_audio_bridge.py b/tests/backend/test_web_audio_bridge.py
index 18fdefaa..1b1cbc3a 100644
--- a/tests/backend/test_web_audio_bridge.py
+++ b/tests/backend/test_web_audio_bridge.py
@@ -1,8 +1,14 @@
import asyncio
+from unittest.mock import MagicMock, patch
import numpy as np
+import pytest
-from meshchatx.src.backend.web_audio_bridge import WebAudioSink, WebAudioSource
+from meshchatx.src.backend.web_audio_bridge import (
+ WebAudioBridge,
+ WebAudioSink,
+ WebAudioSource,
+)
class _DummySink:
@@ -37,3 +43,36 @@ def test_web_audio_sink_encodes_and_sends_bytes():
loop.run_until_complete(asyncio.sleep(0.01))
loop.close()
assert sent, "expected audio bytes to be queued for sending"
+
+
+@pytest.mark.asyncio
+async def test_web_audio_bridge_lazy_loop():
+ """Test that WebAudioBridge retrieves the loop lazily to avoid startup crashes."""
+ mock_tele_mgr = MagicMock()
+ mock_config_mgr = MagicMock()
+
+ # Mock get_event_loop to simulate it not being available during init
+ with patch("asyncio.get_event_loop", side_effect=RuntimeError("No running loop")):
+ bridge = WebAudioBridge(mock_tele_mgr, mock_config_mgr)
+ assert bridge._loop is None
+
+ # Simulate a running loop
+ current_loop = asyncio.get_running_loop()
+ assert bridge.loop == current_loop
+ assert bridge._loop == current_loop
+
+
+def test_web_audio_bridge_asyncutils_fallback():
+ """Test that WebAudioBridge falls back to AsyncUtils.main_loop if no loop is running."""
+ from meshchatx.src.backend.async_utils import AsyncUtils
+
+ mock_loop = MagicMock(spec=asyncio.AbstractEventLoop)
+ AsyncUtils.set_main_loop(mock_loop)
+
+ mock_tele_mgr = MagicMock()
+ mock_config_mgr = MagicMock()
+
+ with patch("asyncio.get_running_loop", side_effect=RuntimeError):
+ bridge = WebAudioBridge(mock_tele_mgr, mock_config_mgr)
+ assert bridge.loop == mock_loop
+ assert bridge._loop == mock_loop
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────